package com.shujia.flink.window

import org.apache.flink.api.scala.ExecutionEnvironment
import org.apache.flink.api.scala._
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.streaming.api.windowing.assigners.{SlidingProcessingTimeWindows, TumblingEventTimeWindows, TumblingProcessingTimeWindows}
import org.apache.flink.streaming.api.windowing.time.Time

object Demo1TImeWindow {
  def main(args: Array[String]): Unit = {

    val env = StreamExecutionEnvironment.getExecutionEnvironment

    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)

    val ds = env.socketTextStream("node1", 8888)


    ds
      .flatMap(_.split(","))
      .map(w => (w, 1))
      .keyBy(_._1)

      /**
        *
        * TumblingEventTimeWindows   事件时间的滚动窗口
        * TumblingProcessingTimeWindows  处理时间的滚动窗口
        *
        *
        * SlidingEventTimeWindows   事件时间的滑动窗口
        * SlidingProcessingTimeWindows   处理时间的滑动窗口
        *
        *
        */
      //滚动窗口 和滚动窗口简写
      //.timeWindow(Time.seconds(10)) //滚动窗口
      //.timeWindow(Time.seconds(10), Time.seconds(2)) //滚动窗口
      .window(SlidingProcessingTimeWindows.of(Time.seconds(5),Time.seconds(1)))
      .reduce((x, y) => (x._1, x._2 + y._2))
      .print()

    env.execute("Demo1Window")

  }
}
